package com.goxplanet;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;

import org.apache.flink.streaming.api.functions.KeyedProcessFunction;

import org.apache.flink.util.Collector;

public class StockTradeProcessFunction extends KeyedProcessFunction<String,StockTrade,StockPosition> {
    private transient ValueState<StockPosition> position;

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) {
        ValueStateDescriptor<StockPosition> descriptor =
                new ValueStateDescriptor<>(
                        "position",
                        StockPosition.class);
        descriptor.setQueryable("abc");
        position = getRuntimeContext().getState(descriptor);
    }
//    @Override
//    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
//        state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
//    }


    @Override
    public void processElement(StockTrade stockTrade, Context context, Collector<StockPosition> collector) throws Exception {
        StockPosition p = position.value();
        if(p == null){
            p = new StockPosition();
            p.ticker = stockTrade.ticker;
            p.volume = stockTrade.volume;
            p.cost = 0.0;
        }else{
            p.volume = p.volume + stockTrade.volume;
            p.cost = p.cost + stockTrade.volume * stockTrade.price;
            p.timestamp = stockTrade.timestamp;
        }

        collector.collect(p);
        position.update(p);
    }
}
